package com.winglinks.mqttclient;

import com.winglinks.config.mqtt.MqttConfig;
import com.winglinks.config.mqtt.PushCallback;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author wangzhong
 * @description: mqtt客户端工厂
 * @date 2024/1/24 16:43
 */
@Component
public class DefaultMqttClientFactory implements MqttClientFactory {
    private static final Logger log = LoggerFactory.getLogger(DefaultMqttClientFactory.class);

    @Autowired
    private MqttConfig mqttConfig;

    @Autowired
    private PushCallback pushCallback;

    Map<String, MyMqttClient> mqttClientRegistry = new ConcurrentHashMap<>();
    /**
     * 根据客户端id查询client
     *
     * @param clientId
     * @return
     */
    @Override
    public MyMqttClient getClient(String clientId,String username,String password) {
        MyMqttClient mqttClient = mqttClientRegistry.get(clientId);
        if(Objects.isNull(mqttClient)){
            mqttClient = this.connect(clientId,username,password);
            if(Objects.nonNull(mqttClient)){
                this.register(clientId,mqttClient);
            }
        }
        return mqttClientRegistry.get(clientId);
    }

    /**
     * 注册client
     *
     * @param clientId
     * @param mqttClient
     */
    @Override
    public void register(String clientId, MyMqttClient mqttClient) {
        mqttClientRegistry.put(clientId,mqttClient);
    }

    /**
     * 连接
     * @param clientID
     * @param username
     * @param password
     * @return
     */

    public MyMqttClient connect(String clientID, String username, String password) {
        MyMqttClient client = null;
        try {
            client = new MyMqttClient(mqttConfig.getHostUrl(), clientID, new MemoryPersistence());
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);
            options.setUserName(username);
            options.setPassword(password.toCharArray());
            options.setConnectionTimeout(mqttConfig.getTimeout());
            options.setKeepAliveInterval(mqttConfig.getKeepalive());
            // automaticReconnect 为 true 表示断线自动重连，但仅仅只是重新连接，并不订阅主题；在 connectComplete 回调函数重新订阅
            options.setAutomaticReconnect(true);
            try {
                //设置回调类
                client.setCallback(pushCallback);
                IMqttToken iMqttToken = client.connectWithResult(options);
                boolean complete = iMqttToken.isComplete();
                log.info("MQTT连接" + (complete ? "成功" : "失败"));
            } catch (Exception e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return client;
    }

    /**
     * 断连
     */
    @PreDestroy
    private void disConnect(){
        for (Map.Entry<String, MyMqttClient> stringMyMqttClientEntry : mqttClientRegistry.entrySet()) {
            try {
                stringMyMqttClientEntry.getValue().disconnect();
            } catch (MqttException e) {
                throw new RuntimeException(e);
            }
        }

    }
}
